package com.citusdata.elven.httpRpc;

import com.citusdata.elven.httpRpc.HttpRpcChannel;
import com.citusdata.elven.util.ConfigManager;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;


public class HttpRpcChannelManager {
  private static final Logger logger = Logger.getLogger(HttpRpcChannelManager.class);
  private static final Configuration config = ConfigManager.getConfig();

  // Mappings from host+port to list of HttpRpcChannels
  private Map usedChannels = new HashMap();
  private Map freeChannels = new HashMap();
  private int usedChannelCount = 0;
  
  // Inverse mappings from HttpRpcChannel to host+port
  private Map leastRecentChannels = new LinkedHashMap();

  private final boolean persistent;
  private final int     maxTotalChannels;

  public HttpRpcChannelManager() {
    persistent = config.getBoolean("HttpRpcChannelManager.Persistent", true);
    maxTotalChannels = config.getInt("HttpRpcChannelManager.MaxTotalChannels", 10);
  }


  public synchronized HttpRpcChannel getChannel(String host, int port) {
    HttpRpcChannel httpChannel = null;

    if (!persistent) {
      httpChannel = new HttpRpcChannel(host, port, persistent);
      return httpChannel;
    }

    if (usedChannelCount >= maxTotalChannels) {
      logger.info("All channels are used, creating non-persistent channel");
      httpChannel = new HttpRpcChannel(host, port, false);
      return httpChannel;
    }

    // Create/choose from available channels, and manage the selected channel
    String endpointId = host + ":" + port;
    int totalChannels = leastRecentChannels.size();

    if (totalChannels < maxTotalChannels) {
      httpChannel = new HttpRpcChannel(host, port, persistent);
    } else {
      // Reuse a previously allocated channel
      List freeList = (List) freeChannels.get(endpointId);
      boolean listEmpty = (freeList == null) || (freeList.isEmpty());

      if (!listEmpty) {
        httpChannel = (HttpRpcChannel) freeList.remove(0);
      } else {
        deleteLeastRecentChannel();
        httpChannel = new HttpRpcChannel(host, port, persistent);
      }
    }

    // Update recency statistics for this channel
    leastRecentChannels.remove(httpChannel);
    leastRecentChannels.put(httpChannel, endpointId);

    // Update usedChannels to track this channel
    List usedList = (List) usedChannels.get(endpointId);
    if (usedList == null) {
      usedList = new LinkedList();
      usedChannels.put(endpointId, usedList);
    }
    usedList.add(httpChannel);
    usedChannelCount++;

    return httpChannel;
  }


  public synchronized void releaseChannel(HttpRpcChannel httpChannel) {
    String endpointId = (String) leastRecentChannels.get(httpChannel);
    if (endpointId == null) {
      return;  // The manager was not tracking this channel; do nothing
    }

    List usedList = (List) usedChannels.get(endpointId);
    assert (usedList != null) : "usedChannel/lru data structure mismatch";

    boolean removed = removeChannelFromList(usedList, httpChannel);
    assert (removed == true) : "usedList/lru data structure mismatch";
    usedChannelCount--;

    if (httpChannel.connected()) {
      // If the channel is still open, keep this channel in the freeList for
      // future re-use.
      List freeList = (List) freeChannels.get(endpointId);
      if (freeList == null) {
        freeList = new LinkedList();
        freeChannels.put(endpointId, freeList);
      }
      freeList.add(httpChannel);

    } else {
      // If the channel has been closed, do not track it anymore.
      leastRecentChannels.remove(httpChannel);
    }
  }


  private void deleteLeastRecentChannel() {
    // LinkedHashMap maintains a doubly-linked list running through all of its
    // entries. This linked list defines the iteration ordering, which in our
    // case is insertion-order.
    for (Iterator iter = leastRecentChannels.entrySet().iterator();
         iter.hasNext(); ) {
 
      Map.Entry pairs = (Map.Entry) iter.next();

      HttpRpcChannel httpChannel = (HttpRpcChannel) pairs.getKey();
      String endpointId = (String) pairs.getValue();

      List freeList = (List) freeChannels.get(endpointId);
      boolean removed = removeChannelFromList(freeList, httpChannel);
      if (removed) {
        iter.remove();
        httpChannel.shutdown();  // Close least recent channel
        return;                  // Removed channel, return
      }
    }
  }


  private boolean removeChannelFromList(List list, HttpRpcChannel channel) {
    if (list == null) {
      return false;
    }

    for (Iterator iter = list.iterator(); iter.hasNext(); ) {
      HttpRpcChannel current = (HttpRpcChannel) iter.next();
      if (current == channel) {
        iter.remove();
        return true;
      }
    }

    return false;
  } 


  public synchronized void shutdown() {
    // Force-close all channels
    for (Iterator iter = leastRecentChannels.keySet().iterator(); iter.hasNext(); ) {
      HttpRpcChannel httpChannel = (HttpRpcChannel) iter.next();
      httpChannel.shutdown();
    }

    leastRecentChannels.clear();
    usedChannels.clear();
    freeChannels.clear();
    usedChannelCount = 0;
  }
}
